package com.zyx.flinkdemo.stream.window.basic;

import com.zyx.flinkdemo.pojo.WordTs;
import com.zyx.flinkdemo.stream.utils.TimeUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

/**
 * @author Yaxi.Zhang
 * @since 2021/6/18 15:49
 * desc: 基于事件时间的滚动窗口案例
 */
public class EventTimeTumblingDemo {

    public static final long TUMBLE_MINUTES = 10L;

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env
                .socketTextStream("localhost", 7777)
                .map(new MapFunction<String, WordTs>() {
                    @Override
                    public WordTs map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WordTs(split[0], Long.parseLong(split[1]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<WordTs>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<WordTs>() {
                            @Override
                            public long extractTimestamp(WordTs element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }))
                .keyBy(WordTs::getWord)
                // 1天滚动窗口, 注意时区
                .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8)))
                // 依据事件时间每10min触发一次计算
                .trigger(ContinuousEventTimeTrigger.of(Time.minutes(10L)))
                .aggregate(new AggregateFunction<WordTs, Tuple2<Integer, Long>, Tuple2<Integer, Long>>() {
                               @Override
                               public Tuple2<Integer, Long> createAccumulator() {
                                   return Tuple2.of(0, -1L);
                               }

                               @Override
                               public Tuple2<Integer, Long> add(WordTs value, Tuple2<Integer, Long> accumulator) {
                                   accumulator.f0 += 1;
                                   accumulator.f1 = Math.max(accumulator.f1, value.getTs());
                                   return accumulator;
                               }

                               @Override
                               public Tuple2<Integer, Long> getResult(Tuple2<Integer, Long> accumulator) {
                                   return accumulator;
                               }

                               @Override
                               public Tuple2<Integer, Long> merge(Tuple2<Integer, Long> a, Tuple2<Integer, Long> b) {
                                   return null;
                               }
                           },
                        new WindowFunction<Tuple2<Integer, Long>, String, String, TimeWindow>() {

                            @Override
                            public void apply(String key, TimeWindow window, Iterable<Tuple2<Integer, Long>> input, Collector<String> out) throws Exception {
                                Tuple2<Integer, Long> wordCount = input.iterator().next();
                                out.collect("++++++++++++ word >> " + key + ", count >> " + wordCount.f0 +
                                        ", triggerTime:" + TimeUtils.getFormatTimeStr(getTriggerTime(wordCount.f1)));
                            }

                            private long getTriggerTime(long maxTs) {
                                return maxTs - (maxTs % TimeUnit.MINUTES.toMillis(TUMBLE_MINUTES)) + TimeUnit.MINUTES.toMillis(TUMBLE_MINUTES);
                            }
                        })
                .print("wordCount >>>");
        /*
            结果分析
            发送
                zhangsan,1623913860000                          2021-06-17 15:11:00
                zhangsan,1623914400000                          2021-06-17 15:20:00
                zhangsan,1623914400001                          2021-06-17 15:20:00.001
                zhangsan,1623914401000                          2021-06-17 15:20:01
                zhangsan,1623914460000                          2021-06-17 15:21:00
                zhangsan,1623914700000                          2021-06-17 15:25:00
                zhangsan,1623915600000                          2021-06-17 15:40:00
                zhangsan,1623916200000                          2021-06-17 15:50:00
                zhangsan,1623945600000                          2021-06-18 00:00:00
            结果
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 3, triggerTime:2021-06-17 15:30:00 第3条触发
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 7, triggerTime:2021-06-17 15:50:00 第7条触发
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00 第8条触发
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00 后面都是第9条触发
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
                wordCount >>>> ++++++++++++ word >> zhangsan, count >> 8, triggerTime:2021-06-17 16:00:00
            可以看到使用ContinuousEventTimeTrigger时, 只有在超过interval整数倍1ms时才会触发计算, 当到达窗口关闭时间时会一次触发49条,
            下午3点50到凌晨还有8个小时10分钟, 每个小时触发6次, 一共8*6+1=49次
         */

        env.execute("word_count");
    }
}
